package trouger.dmt;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;


class JoinableKey
implements WritableComparable<JoinableKey>{

	/**
	 * zero-based source index
	 */
	protected int sourceIndex;
	/**
	 * key-column values
	 */
	protected String[] keyColumn;
	protected int columnCount;
	protected int hashCode_precomputed;
	
	public JoinableKey(){
		//this method is only declared for use in JobConf.getOutputKeyComparator
	}
	
	public JoinableKey(int sourceIndex, String[] keyColumn){
		this.sourceIndex = sourceIndex;
		this.keyColumn = keyColumn;
		this.columnCount = keyColumn.length;
		if (columnCount == 1){
			hashCode_precomputed = keyColumn[0].hashCode();
		}else{
			int power31 = 1;
			hashCode_precomputed = 0;
			for (int i = 0; i < columnCount; i++){
				hashCode_precomputed += keyColumn[i].hashCode() * power31;
				power31 *= 31;
			}
		}
	}
	
	@Override
	public void readFields(DataInput in) throws IOException {
		sourceIndex = in.readInt();
		columnCount = in.readInt();
		keyColumn = new String[columnCount];
		for (int i = 0; i < columnCount; i++){
			keyColumn[i] = in.readUTF();
		}
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeInt(sourceIndex);
		out.writeInt(columnCount);
		for (int i = 0; i < columnCount; i++){
			out.writeUTF(keyColumn[i]);
		}
	}

	@Override
	public int compareTo(JoinableKey o) {
		int n = this.columnCount;
		if (n != o.columnCount){
			throw new IllegalStateException("two join keys not match in column-count");
		}
		for (int i = 0; i < n; i++){
			int t = this.keyColumn[i].compareTo(o.keyColumn[i]);
			if (t != 0) return t;
		}
		return this.sourceIndex - o.sourceIndex;
	}
	
	@Override
	public int hashCode(){
		return this.hashCode_precomputed;
	}
}

class JoinableValue implements Writable{

	/**
	 * zero-based source index
	 */
	protected int sourceIndex;
	protected int columnCount;
	/**
	 * value column values
	 */
	protected String[] valueColumn;
	
	public JoinableValue(){
		//this method is only declared for use in WritableDeserializer.deserialize
	}
	
	public JoinableValue(int sourceIndex, String[] valueColumn){
		this.sourceIndex = sourceIndex;
		this.valueColumn = valueColumn;
		this.columnCount = valueColumn.length;
	}
	
	@Override
	public void readFields(DataInput in) throws IOException {
		sourceIndex = in.readInt();
		columnCount = in.readInt();
		valueColumn = new String[columnCount];
		for (int i = 0; i < columnCount; i++){
			valueColumn[i] = in.readUTF();
		}
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeInt(sourceIndex);
		out.writeInt(columnCount);
		for (int i = 0; i < columnCount; i++){
			out.writeUTF(valueColumn[i]);
		}
	}
	
}

class JoinableGroupingComparator extends WritableComparator {

	protected JoinableGroupingComparator() {
		super(JoinableKey.class);
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public int compare(WritableComparable a, WritableComparable b) {
		JoinableKey key1 = (JoinableKey)a;
		JoinableKey key2 = (JoinableKey)b;
		int n = key1.columnCount;
		if (n != key2.columnCount){
			throw new IllegalStateException("two join keys not match in column-count");
		}
		for (int i = 0; i < n; i++){
			int t = key1.keyColumn[i].compareTo(key2.keyColumn[i]);
			if (t != 0) return t;
		}
		return 0;
	}
}

class JoinableMapperEntity
implements MapperEntity<JoinableKey, JoinableValue>{

	//these indices are one-based.
	protected int[][] keyColumnList;
	protected int[][] valueColumnList;
	protected int[] maxColumnIndexOfSource;
	
	protected JoinableKey key;
	protected JoinableValue value;
	
	public void configure(int[][] keyColumnList, int[][] valueColumnList){
		this.keyColumnList = keyColumnList;
		this.valueColumnList = valueColumnList;
		int sourceCount = keyColumnList.length;
		maxColumnIndexOfSource = new int[sourceCount];
		for (int sourceIndex = 0; sourceIndex < sourceCount; sourceIndex++){
			int[] keyIndices = keyColumnList[sourceIndex];
			int[] valueIndices = valueColumnList[sourceIndex];
			int maxIndex = -1;
			for (int i = 0; i < keyIndices.length; i++){
				maxIndex = Math.max(maxIndex, keyIndices[i]);
			}
			for (int i = 0; i < valueIndices.length; i++){
				maxIndex = Math.max(maxIndex, valueIndices[i]);
			}
			maxColumnIndexOfSource[sourceIndex] = maxIndex;
		}
	}
	
	@Override
	public boolean acceptLine(int sourceIndex, String[] segment) {
		key = null;
		value = null;
		int[] keyIndices = keyColumnList[sourceIndex];
		int[] valueIndices = valueColumnList[sourceIndex];

		if (maxColumnIndexOfSource[sourceIndex] >= segment.length){
			System.err.println("column index exceeds column number.");
			System.err.println("\tthe total number of columns from source table " + (sourceIndex + 1) + " is " + (segment.length - 1));
			System.err.println("\twhile column index " + (maxColumnIndexOfSource[sourceIndex]) + " is specified in either key or value fields.");
			return false;
		}
		String[] keyColumn = new String[keyIndices.length];
		for (int i = 0; i < keyIndices.length; i++){
			keyColumn[i] = segment[keyIndices[i]];
		}
		key = new JoinableKey(sourceIndex, keyColumn);
		String[] valueColumn = new String[valueIndices.length];
		for (int i = 0; i < valueIndices.length; i++){
			valueColumn[i] = segment[valueIndices[i]];
		}
		value = new JoinableValue(sourceIndex, valueColumn);
		return true;
	}

	@Override
	public JoinableKey getKey() {
		return key;
	}

	@Override
	public JoinableValue getValue() {
		return value;
	}

	
}

class JoinableMapper 
extends BaseMapper<JoinableMapperEntity, JoinableKey, JoinableValue> {
	
	@Override 
	public void configure(JobConf conf){
		super.configure(conf);
		this.mapperEntity.configure(keyColumnList,  valueColumnList);
	}
}
